//==============================================================================================================================================================
// Author(s): Wei Jiang
// Updated by Wei 2013/08/11
// This module will be droppped
// Class used for tcp connection
//==============================================================================================================================================================


#include "tcp_connection.h"
#include <boost/serialization/vector.hpp>

using namespace std;


/*
 * session class
 *
 * */
  sessionForBroker::sessionForBroker(boost::asio::io_service& io_service)
    : socket_(io_service)
  {
  }

  tcp::socket& sessionForBroker::socket()
  {
    return socket_;
  }

  void sessionForBroker::welcome(string queryString){
    cout << "Welcome" << endl;
    cout << "queryString is:" << queryString << endl;
  }

  string sessionForBroker::getReturningResult(){
    string resultString = "";
    //cout << "bytes_transferred_:" << bytes_transferred_ << endl;
    //cout << "returnResult_:" << returnResult_ << endl;
    for(size_t i=0; i < bytes_transferred_; i++){
        resultString += returnResult_[i];
    }
    return resultString;

  }

  void sessionForBroker::addSpecificQueryToTCPSession(int number, string queryContent){
    query q;
    q.queryNumber = number;
    q.queryContent = queryContent;
    queries_loaded_.push_back(q);

  }

  int sessionForBroker::getQueryLogSize(){
    return queries_loaded_.size();
  }

  void sessionForBroker::startQueryOnThisMachine(vector<pair<int, string> > &queries)
  {
    cout << "in new version startQueryOnThisMachine" << endl;

    /*
    cout << queries->size() << endl;

    for (int i = 0; i < static_cast<int> (queries->size()); ++i) {
      //cout << queries[i]first << ":" << queries[i]->second << endl;
    }
    */

  }

  void sessionForBroker::startQueryOnThisMachine(vector<string> &queries)
  {
    cout << "in new version startQueryOnThisMachine try3(currently, do nothing)" << endl;


    //cout << queries.size() << endl;

    //for (int i = 0; i < queries.size(); ++i) {}
  }

  void sessionForBroker::startQueryOnThisMachineWithOneQuery(string queryString)
  {

    memcpy(data_,queryString.c_str(),queryString.size());

    socket_.async_write_some(boost::asio::buffer(data_, queryString.size()),boost::bind(&sessionForBroker::handle_write, this,boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));


  }

  void sessionForBroker::startQueryOnThisMachineWithQueriesInside(){
    //cout << "start Query On This Machine With Queries Inside(new beginning)" << endl;

    async_write(queries_loaded_,boost::bind(&sessionForBroker::handle_write_for_queries, this,boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
  }

  void sessionForBroker::handle_read_for_queries(const boost::system::error_code& error)
  {
    if (!error)
    {
      //cout << "Received The Whole Queries Result Log" << endl;
      //cout << "Size of results:" << queries_with_results_.size() << endl;

      for (int i = 0; i < static_cast<int> (queries_with_results_.size()); ++i) {
        cout << queries_with_results_[i].queryContent << " with " << queries_with_results_[i].results.size() << " results" << endl;
      }

      cout << "this is the test part(2012/02/28), the 1 result is:" << queries_with_results_[0].results[0].url << endl;

      cout << "Search Job Finished by " << socket_.remote_endpoint() << endl;

      // Each time when finishing the job(one write and one read), it can just reset queries_loaded_ and the queries_with_results_ vectors.
      queries_loaded_.clear();
      queries_with_results_.clear();
    }
    else
    {
        cout << "error:" << error.message() << endl;
      delete this;
    }
  }

  void sessionForBroker::handle_read(const boost::system::error_code& error, size_t bytes_transferred)
  {
    if (!error)
    {
      //cout << "Receive data from:" << socket_.remote_endpoint()<<endl;
      cout << "Received Begin:" << endl;
      cout.write(returnResult_, bytes_transferred);
      bytes_transferred_ = bytes_transferred;
      cout << "Received End." << endl;

      /*
      boost::asio::async_write(socket_,
          boost::asio::buffer(data_, bytes_transferred),
          boost::bind(&session::handle_write, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
      */
    }
    else
    {
        cout << "error:" << error.message() << endl;
      delete this;
    }
  }

  void sessionForBroker::handle_write_for_queries(const boost::system::error_code& error, size_t bytes_transferred)
  {

    if (!error)
    {
      cout << bytes_transferred << " bytes transferred(sent)." << endl;

      async_read(queries_with_results_,
          boost::bind(&sessionForBroker::handle_read_for_queries, this,
            boost::asio::placeholders::error));
    }
    else
    {
        cout << "error:" << error.message() << endl;
      delete this;
    }
  }

  void sessionForBroker::handle_write(const boost::system::error_code& error, size_t bytes_transferred)
  {
    //cout << "handle_write called" << endl;
    if (!error)
    {

      //cout << "Send data to:" << socket_.remote_endpoint()<<endl;
      cout << "Sent Begin:" << endl;
      cout.write(data_, bytes_transferred);
      cout << "Sent End." << endl;


      socket_.async_read_some(boost::asio::buffer(returnResult_, max_length),
          boost::bind(&sessionForBroker::handle_read, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
    }
    else
    {
        cout << "error:" << error.message() << endl;
      delete this;
    }
  }

/*
 * class sessionOriginal
 *
 * */



    sessionOriginal::sessionOriginal(boost::asio::io_service& io_service)
      : socket_(io_service)
    {
    }

    tcp::socket& sessionOriginal::socket()
    {
      return socket_;
    }

    void sessionOriginal::start(string queryString)
    {
      boost::asio::async_write(socket_,
          boost::asio::buffer(queryString, queryString.size()),
          boost::bind(&sessionOriginal::handle_write, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
      /*
      socket_.async_read_some(boost::asio::buffer(data_, max_length),
          boost::bind(&sessionOriginal::handle_read, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
            */
    }


    void sessionOriginal::handle_read(const boost::system::error_code& error,
        size_t bytes_transferred)
    {
      if (!error)
      {
        cout << "Received:";
        cout.write(data_, bytes_transferred);
        cout << endl;
        boost::asio::async_write(socket_,
            boost::asio::buffer(data_, bytes_transferred),
            boost::bind(&sessionOriginal::handle_write, this,
              boost::asio::placeholders::error,
              boost::asio::placeholders::bytes_transferred));
      }
      else
      {
        delete this;
      }
    }

    void sessionOriginal::handle_write(const boost::system::error_code& error, size_t bytes_transferred)
    {
      if (!error)
      {
        cout << "Sent:";
        cout.write(data_,bytes_transferred);
        cout << endl;
        socket_.async_read_some(boost::asio::buffer(data_, max_length),
            boost::bind(&sessionOriginal::handle_read, this,
              boost::asio::placeholders::error,
              boost::asio::placeholders::bytes_transferred));
      }
      else
      {
        delete this;
      }
    }



    /*
     * class sessionForSlave
     *
     * */
/*
      sessionForSlave::sessionForSlave(boost::asio::io_service& io_service)
        : socket_(io_service)
      {
      }

      tcp::socket& sessionForSlave::socket()
      {
        return socket_;
      }

      void sessionForSlave::start()
      {
        cout << "Get connection from:" <<socket_.remote_endpoint() << endl;
        socket_.async_read_some(boost::asio::buffer(data_, max_length),
            boost::bind(&sessionForSlave::handle_read, this,
              boost::asio::placeholders::error,
              boost::asio::placeholders::bytes_transferred));
      }


      void sessionForSlave::handle_read(const boost::system::error_code& error,
          size_t bytes_transferred)
      {
        if (!error)
        {
          cout << "Received:";
          cout.write(data_, bytes_transferred);
          cout << endl;

          // Call the ExecuteQuery() to get the result.
          string sendBackString;
          //DistributedSlave ds;

          // Right now, I am just hardcoded.
          //sendBackString = ds.ExecuteQuery("edu", 0);

          boost::asio::async_write(socket_,
              boost::asio::buffer(data_, bytes_transferred),
              boost::bind(&sessionForSlave::handle_write, this,
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));
        }
        else
        {
          cout << "Close connection:" << socket_.remote_endpoint() << " with error message:"<< error.message() << endl;
          delete this;
        }
      }

      void sessionForSlave::handle_write(const boost::system::error_code& error, size_t bytes_transferred)
      {
        if (!error)
        {
          cout << "Sent:";
          cout.write(data_,bytes_transferred);
          cout << endl;
          socket_.async_read_some(boost::asio::buffer(data_, max_length),
              boost::bind(&sessionForSlave::handle_read, this,
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));
        }
        else
        {
          cout << "Close connection:" << socket_.remote_endpoint() << " with error message:"<< error.message()<< endl;
          delete this;
        }
      }
*/

    /*
     * class tcp_server
     *
     * */

/*
      tcp_server::tcp_server(boost::asio::io_service& io_service, short port)
        : io_service_(io_service),
          acceptor_(io_service, tcp::endpoint(tcp::v4(), port))
      {
        start_accept();
      }

      tcp_server::tcp_server(boost::asio::io_service& io_service, short port, DistributedSlave* DistributedSlavePointer)
      : io_service_(io_service),
        acceptor_(io_service, tcp::endpoint(tcp::v4(), port))
      {
        start_accept();
      }

      void tcp_server::start_accept()
      {
        sessionForSlave* new_session = new sessionForSlave(io_service_);
        acceptor_.async_accept(new_session->socket(),
            boost::bind(&tcp_server::handle_accept, this, new_session,
              boost::asio::placeholders::error));
      }

      void tcp_server::handle_accept(sessionForSlave* new_session,
          const boost::system::error_code& error)
      {

        if (!error)
        {
          new_session->start();
        }
        else
        {
          delete new_session;
        }

        start_accept();
      }
      */
